package com.we.risk.behavior.loan;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.sql.Timestamp;

import static com.we.risk.behavior.repay.CommonVar.ACCUSERKEY;
import static com.we.risk.behavior.repay.CommonVar.USERKEY;

public class TableHfqLoanCenterRiskRecord {
    public static final String KFKTABLE = "fdm.hhl01_hfq_loancenter_risk_record";
    public static final String QUERYSQL =
            "select user_key, loan_id, " +
                    "loan_line, " +
                    "avail_line as loan_avail_line, " +
                    "loan_rate, " +
                    "`period` as apply_period, " +
                    "amount as apply_amount, " +
                    "create_time as apply_time, " +
                    "avail_periods as loan_period, " +
                    "uid " +
                    "from " + KFKTABLE
                    + " where user_key is not null " +
                    "and create_time is not null " +
                    "and avail_periods is not null";

    public String user_key;
    public Long loan_id;
    public BigDecimal loan_line;
    public BigDecimal loan_avail_line;
    public BigDecimal loan_rate;
    public int apply_period;
    public BigDecimal apply_amount;
    public Timestamp apply_time;
    public String loan_period;
    public Long uid;


    public static Logger LOG = LoggerFactory.getLogger(TableHfqLoanCenterRiskRecord.class);

    public TableHfqLoanCenterRiskRecord() {
    }

    public TableHfqLoanCenterRiskRecord(String user_key, Long loan_id, BigDecimal loan_line,
                                        BigDecimal loan_avail_line, BigDecimal loan_rate,
                                        int apply_period, BigDecimal apply_amount,
                                        Timestamp apply_time, String loan_period, Long uid) {
        this.user_key = user_key;
        this.loan_id = loan_id;
        this.loan_line = loan_line;
        this.loan_avail_line = loan_avail_line;
        this.loan_rate = loan_rate;
        this.apply_period = apply_period;
        this.apply_amount = apply_amount;
        this.apply_time = apply_time;
        this.loan_period = loan_period;
        this.uid = uid;
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableHfqLoanCenterRiskRecord.QUERYSQL);
        DataStream<Tuple2<Boolean, TableHfqLoanCenterRiskRecord>> inputStream =
                tableEnv.toRetractStream(tableRes, TableHfqLoanCenterRiskRecord.class);

        SingleOutputStreamOperator<String> out = inputStream.filter(new FilterFunction<Tuple2<Boolean, TableHfqLoanCenterRiskRecord>>() {
            @Override
            public boolean filter(Tuple2<Boolean, TableHfqLoanCenterRiskRecord> value) throws Exception {
                return value.f0;
            }
        }).uid("TableHfqLoanCenterRiskRecord-filter")
        .flatMap(new RichFlatMapFunction<Tuple2<Boolean, TableHfqLoanCenterRiskRecord>, String>() {
            @Override
            public void flatMap(Tuple2<Boolean, TableHfqLoanCenterRiskRecord> value, Collector<String> out) throws Exception {
                TableHfqLoanCenterRiskRecord input = value.f1;

                String loan_period = input.getLoan_period();
                JSONArray loan_periods = JSONArray.parseArray(loan_period);
                int size = loan_periods.size();
                int max_loan_period = 0;
                for (int i = 0; i < size; i++) {
                    JSONObject element = loan_periods.getJSONObject(i);
                    int period = element.getIntValue("period");
                    max_loan_period = (max_loan_period > period ? max_loan_period : period);
                }
                out.collect(consResultJson(input, (max_loan_period == 0 ? null : max_loan_period)));
            }

            private String consResultJson(TableHfqLoanCenterRiskRecord record, Integer max_loan_period) {
                JSONObject resJson = new JSONObject();
                resJson.put("user_key", record.getUser_key());
                resJson.put("loan_id", record.getLoan_id()+"");
                resJson.put("loan_line", record.getLoan_line());
                resJson.put("loan_avail_line", record.getLoan_avail_line());
                resJson.put("loan_rate", record.getLoan_rate());
                resJson.put("apply_period", record.getApply_period());
                resJson.put("apply_amount", record.getApply_amount());
                resJson.put("apply_time", record.getApply_time());
                resJson.put("loan_period", max_loan_period);
                resJson.put("uid", record.getUid());
                return resJson.toString();
            }
        });

        return out;
    }

    public Long getUid() {
        return uid;
    }

    public String getUser_key() {
        return user_key;
    }

    public Long getLoan_id() {
        return loan_id;
    }

    public BigDecimal getLoan_line() {
        return loan_line;
    }

    public Timestamp getApply_time() {
        return apply_time;
    }

    public BigDecimal getLoan_avail_line() {
        return loan_avail_line;
    }

    public BigDecimal getLoan_rate() {
        return loan_rate;
    }

    public int getApply_period() {
        return apply_period;
    }

    public BigDecimal getApply_amount() {
        return apply_amount;
    }

    public String getLoan_period() {
        return loan_period;
    }
}
